package com.bigdata.es.kafkaproducer;

import com.alibaba.fastjson.JSONObject;
import com.bigdata.es.conf.KafkaProperties;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.Collections;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutionException;

public class KafkaProducerExample {

    private static final Logger logger = LogManager.getLogger(KafkaProducerExample.class);
    //topic
    private static String topicname = KafkaProperties.getTopic();
    //用户id
    private static long id = 1L;

    public static void main(String[] args) {
        // Create topic
        try{
            createTopic();
        } catch (ExecutionException |InterruptedException e){
            logger.error("An error occurred.", e);
        }
        Random random = new Random();
        String[] names = new String[]{"Sean", "Bill", "Thomas", "Lucy", "Anna"};
        Address[] addressArray = new Address[5];
        addressArray[0]=new Address(1, "China", "Beijing");
        addressArray[1]=new Address(2, "China", "Shanghai");
        addressArray[2]=new Address(3, "US", "New York");
        addressArray[3]=new Address(4, "UK", "London");
        addressArray[4]=new Address(5, "Japan", "Tokyo");
        final Producer<String, String> producer = KafkaProperties.createProducer();
        int PRODUCER_INTERVAL = 200;
        try{
            while (true){
                String uuid = UUID.randomUUID().toString();
                int nameIndex = random.nextInt(5);
                int addressIndex = random.nextInt(5);
                Customer customer = new Customer();
                customer.setName(names[nameIndex]);
                customer.setId(id++);
                customer.setAddress(addressArray[addressIndex]);
                customer.setDescription("" + names[nameIndex] + "来自"
                        + addressArray[addressIndex].getCity() + ", " +
                        addressArray[addressIndex].getCountry());
                String customerJson = JSONObject.toJSONString(customer);
                ProducerRecord<String, String> record = new ProducerRecord<>(topicname, uuid, customerJson);
                RecordMetadata metadata = producer.send(record).get();

                logger.info("Sent ({}, {}) to topic {} @ {}.", uuid, customerJson, topicname, metadata.timestamp());

                Thread.sleep(PRODUCER_INTERVAL);
            }
        } catch (InterruptedException | ExecutionException e) {
            logger.error("An error occurred.", e);
        } finally {
            producer.flush();
            producer.close();
        }
    }

    private static void createTopic() throws ExecutionException, InterruptedException {
        //分区数
        int partition = KafkaProperties.getPartition();
        //副本数
        short replication = KafkaProperties.getReplication();
        AdminClient admin = KafkaProperties.createAdminClient();
        ListTopicsResult result = admin.listTopics();
        Set<String> resultSet = result.names().get();
        System.out.println(resultSet);
        if(!resultSet.contains(topicname)){
            NewTopic topic = new NewTopic(topicname, partition, replication);
            admin.createTopics(Collections.singletonList(topic));
            Thread.sleep(500);
            logger.info("Topic {} created", topicname);
        }else {
            Thread.sleep(500);
            logger.info("Topic {} already exists", topicname);
        }
        admin.close();
    }
}
